package cn.bby.query.server.infrastructure.impl.event;

import cn.bby.jtk.base.task.AsyncUtil;
import cn.bby.query.server.application.manager.DataChangeEventManager;
import cn.bby.query.server.domain.service.dto.DataChangeEvent;
import cn.bby.query.server.infrastructure.impl.event.model.Payload;
import cn.bby.query.server.infrastructure.impl.event.model.Source;
import com.alibaba.fastjson2.JSON;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Locale;
import java.util.function.Consumer;

/**
 * @author wangbenyin
 * @version 1.0
 * @date 2025/06/20
 * @since 17
 */
@Component
public class CloudStreamChangeEventManager extends DataChangeEventManager {
    @Autowired
    private StreamBridge streamBridge;
    @Value("${query.event.max-consumer-thread:2}")
    private int maxConsumerThread = 2;

    @Override
    public void sendEvent(List<String> payloads) {
        streamBridge.send("dataChangeEvent-out-0", payloads);
    }

    @Bean
    public Consumer<List<String>> dataChangeEvent() {
        return payloads -> AsyncUtil.runAsync(() -> {
            List<DataChangeEvent> events = payloads.stream().map(payloadStr -> {
                        Payload payload = JSON.parseObject(payloadStr, Payload.class);
                        Source source = payload.source;
                        if (source == null || payload.op == null || payload.after == null && payload.before == null) {
                            return null;
                        }
                        DataChangeEvent event = new DataChangeEvent();
                        event.setTableFullName("%s.%s.%s".formatted(source.getDb(), source.getSchema(), source.getTable()));
                        event.setOpType(payload.op.toUpperCase(Locale.ROOT));
                        event.setBefore(payload.before);
                        event.setAfter(payload.after);
                        return event;
                    }).filter(e -> e != null && (e.getAfter() != null || e.getBefore() != null))
                    .toList();
            handleEvent(events);
        });
    }

}
